package com.atguigu.sink;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import com.mysql.cj.jdbc.MysqlXADataSource;
import org.apache.flink.connector.jdbc.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.function.SerializableSupplier;

import javax.sql.XADataSource;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/15 11:01
 *
 *  JdbcSink
 *      1. 常规写入
 *      2. exactilyOnceSink() 支持分布式写入，支持分布式事务， 支持EOS
 *  XASink：
 *      注意事项：
 *          1. 先明确当前使用的数据是否支持XA事务
 *          2. 如果使用的是Mysql或者PostgreSQL，在一个链接上只支持一个XA事务
 *              JdbcExactlyOnceOptions.builder()
 *              .witTransactionPerConnection(true)
 *              .build();
 *          3. 如果使用MySQL XA ，需要提前MySQL的用户授权 XA_RECOVER_AdMIN
 *          4. 如果重试次数只能是 0 .JdbcExeccutionOptions.maxRetries == 0
 *          5. 开启flink检查点
 */
public class Flink05_JdbcXASink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        ds.print("input");

        //将流中的数据写入到Mysql
        SinkFunction<Event> jdbcSink = JdbcSink.<Event>exactlyOnceSink(
                //"insert into clicks1 (user ,url ,ts ) values(?,?,?)",
                "replace into clicks(user, url, ts ) values(?,?,?)",
                new JdbcStatementBuilder<Event>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                        preparedStatement.setString(1, event.getUser());
                        preparedStatement.setString(2, event.getUrl());
                        preparedStatement.setLong(3, event.getTs());
                    }
                },

                JdbcExecutionOptions.builder()
                        .withBatchSize(5)  // 批写大小
                        .withBatchIntervalMs(10000)  // 批次间隔时间， 超过指定时间也要写入
                        .withMaxRetries(0)  //重试次数
                        .build(),

                JdbcExactlyOnceOptions.builder()
                        .withTransactionPerConnection(true)
                        .build(),

                new SerializableSupplier<XADataSource>() {
                    @Override
                    public XADataSource get() {
                        //按照使用的数据库来创建对应的XADataSource
                        MysqlXADataSource mysqlXADataSource = new MysqlXADataSource();
                        mysqlXADataSource.setUrl("jdbc:mysql://hadoop102:3306/test");
                        mysqlXADataSource.setUser("root");
                        mysqlXADataSource.setPassword("000000");
                        return mysqlXADataSource;
                    }
                }
        );

        //旧的API
        ds.addSink(jdbcSink) ;

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
}
}
